package io.wen.bd.s3m3.zookeeper;

import com.alibaba.druid.pool.DruidDataSource;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import javax.sql.DataSource;
import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Logger;

/**
 * 动态数据源
 * 将数据库连接信息交给Zookeeper配置中心管理，项目启动时，从Zookeeper进行MySQL配置参数的拉取。
 */
public class DynamicDataSource implements DataSource {
    private static final ObjectMapper objectMapper = new ObjectMapper();
    // 所有配置项解析存储在configMap中
    private Map<String, String> configMap = new HashMap<>();

    private ZkClient zkClient;
    private String configPath;
    // 代理数据源，使用Druid
    private DruidDataSource dataSource;

    /**
     * 初始化动态数据源
     *
     * @param quorums    zookeeper集群地址 ip:port,...
     * @param configPath 数据源配置存放路径
     */
    public DynamicDataSource(String quorums, String configPath) {
        // 1. 初始化zkClient，配置路径
        zkClient = new ZkClient(quorums);
        this.configPath = configPath;
        // 2. 读取zookeeper中的配置
        readConfigFromZk();
        // 3. 设置数据源
        setupDataSource();
        // 4. 监听zookeeper配置节点的数据变化
        subscribeZk();
    }

    /**
     * 读取zookeeper中存储的配置信息，存储到configMap中
     * 存储在zookeeper中的数据格式：
     * {"url":"jdbc:mysql://hadoop03:3306/mysql","driverClassName":"com.mysql.jdbc.Driver","username":"hive","password":"hive"}
     */
    private void readConfigFromZk() {
        try {
            String rawConfig = zkClient.readData(configPath);
            final Map<String, String> rawConfigMap;
            rawConfigMap = objectMapper.readValue(rawConfig, Map.class);
            configMap.putAll(rawConfigMap);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    /**
     * 设置代理 DataSource，使用Druid
     */
    private void setupDataSource() {
        DruidDataSource proxyDataSource = new DruidDataSource();
        proxyDataSource.setUrl(configMap.get("url"));
        proxyDataSource.setDriverClassName(configMap.get("driverClassName"));
        proxyDataSource.setUsername(configMap.get("username"));
        proxyDataSource.setPassword(configMap.get("password"));
        this.dataSource = proxyDataSource;
    }

    /**
     * 监听zookeeper配置路径的数据变化
     */
    private void subscribeZk() {
        zkClient.subscribeDataChanges(configPath, new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                readConfigFromZk();
                // 切换新数据源，关闭老数据源
                DruidDataSource oldDataSource = DynamicDataSource.this.dataSource;
                setupDataSource();
                oldDataSource.close();
            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
            }
        });
    }

    public Map<String, String> getConfigMap() {
        return configMap;
    }

    @Override
    public Connection getConnection() throws SQLException {
        return dataSource.getConnection();
    }

    @Override
    public Connection getConnection(String username, String password) throws SQLException {
        return dataSource.getConnection(username, password);
    }

    @Override
    public <T> T unwrap(Class<T> iface) throws SQLException {
        return null;
    }

    @Override
    public boolean isWrapperFor(Class<?> iface) throws SQLException {
        return false;
    }

    @Override
    public PrintWriter getLogWriter() throws SQLException {
        return null;
    }

    @Override
    public void setLogWriter(PrintWriter out) throws SQLException {

    }

    @Override
    public void setLoginTimeout(int seconds) throws SQLException {

    }

    @Override
    public int getLoginTimeout() throws SQLException {
        return 0;
    }

    @Override
    public Logger getParentLogger() throws SQLFeatureNotSupportedException {
        return null;
    }

}
